package com.lefeng.cockpit.datax.thread;

import com.lefeng.cockpit.datax.conf.JobAdminConfig;
import com.lefeng.cockpit.datax.trigger.JobTrigger;
import com.lefeng.cockpit.datax.trigger.TriggerTypeEnum;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Manages the thread pools for job triggering.
 * It provides methods to start and stop the thread pools,
 * and to add job triggers to the appropriate thread pool based on job timeout statistics.
 */
public class JobTriggerPoolHelper {
    private static final Logger logger = LoggerFactory.getLogger(JobTriggerPoolHelper.class);

    // Thread pools for fast and slow triggers
    private ThreadPoolExecutor fastTriggerPool;
    private ThreadPoolExecutor slowTriggerPool;

    // Job timeout statistics
    private volatile long lastMinTime = System.currentTimeMillis() / 60000; // Last minute timestamp
    private final ConcurrentMap<Integer, AtomicInteger> jobTimeoutCountMap = new ConcurrentHashMap<>();

    // Singleton instance
    private static final JobTriggerPoolHelper INSTANCE = new JobTriggerPoolHelper();

    private JobTriggerPoolHelper() {
    }

    /**
     * Gets the singleton instance of JobTriggerPoolHelper.
     *
     * @return The singleton instance.
     */
    public static JobTriggerPoolHelper getInstance() {
        return INSTANCE;
    }

    /**
     * Starts the thread pools for job triggering.
     */
    public void start() {
        fastTriggerPool = createThreadPool(
                JobAdminConfig.getAdminConfig().getTriggerPoolFastMax(),
                "fastTriggerPool"
        );

        slowTriggerPool = createThreadPool(
                JobAdminConfig.getAdminConfig().getTriggerPoolSlowMax(),
                "slowTriggerPool"
        );
    }

    /**
     * Creates a ThreadPoolExecutor with the specified maximum pool size and pool name.
     *
     * @param maxPoolSize The maximum number of threads in the pool.
     * @param poolName    The name of the pool.
     * @return The created ThreadPoolExecutor.
     */
    private ThreadPoolExecutor createThreadPool(int maxPoolSize, String poolName) {
        return new ThreadPoolExecutor(
                10,
                maxPoolSize,
                60L,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(1000),
                r -> new Thread(r, "JobTriggerPoolHelper-" + poolName + "-" + r.hashCode())
        );
    }

    /**
     * Stops the thread pools and logs the shutdown status.
     */
    private void stop() {
        fastTriggerPool.shutdownNow();
        slowTriggerPool.shutdownNow();
        logger.info("Job trigger thread pools shutdown successfully.");
    }

    /**
     * Adds a trigger task to the appropriate thread pool based on job timeout statistics.
     *
     * @param jobId                 The ID of the job to trigger.
     * @param triggerType           The type of trigger.
     * @param failRetryCount        The number of retry attempts.
     * @param executorShardingParam Sharding parameters for the executor.
     * @param executorParam         Parameters for the executor.
     */
    private void addTrigger(int jobId, TriggerTypeEnum triggerType, int failRetryCount, String executorShardingParam, String executorParam) {
        ThreadPoolExecutor targetPool = getTargetThreadPool(jobId);

        targetPool.execute(() -> {
            long start = System.currentTimeMillis();
            try {
                JobTrigger.trigger(jobId, triggerType, failRetryCount, executorShardingParam, executorParam);
            } catch (Exception e) {
                logger.error("Error triggering job {}: {}", jobId, e.getMessage(), e);
            } finally {
                updateTimeoutStatistics(jobId, start);
            }
        });
    }

    /**
     * Determines the appropriate thread pool based on job timeout statistics.
     *
     * @param jobId The ID of the job.
     * @return The target ThreadPoolExecutor.
     */
    private ThreadPoolExecutor getTargetThreadPool(int jobId) {
        AtomicInteger jobTimeoutCount = jobTimeoutCountMap.get(jobId);
        if (jobTimeoutCount != null && jobTimeoutCount.get() > 10) { // 10 timeouts in 1 minute
            return slowTriggerPool;
        }
        return fastTriggerPool;
    }

    /**
     * Updates the job timeout statistics based on the duration of the job trigger.
     *
     * @param jobId The ID of the job.
     * @param start The start time of the job trigger.
     */
    private void updateTimeoutStatistics(int jobId, long start) {
        long currentMinTime = System.currentTimeMillis() / 60000;
        if (lastMinTime != currentMinTime) {
            lastMinTime = currentMinTime;
            jobTimeoutCountMap.clear();
        }

        long cost = System.currentTimeMillis() - start;
        if (cost > 500) { // Timeout threshold: 500ms
            jobTimeoutCountMap.computeIfAbsent(jobId, k -> new AtomicInteger(0)).incrementAndGet();
        }
    }

    /**
     * Starts the JobTriggerPoolHelper.
     */
    public static void toStart() {
        INSTANCE.start();
    }

    /**
     * Stops the JobTriggerPoolHelper.
     */
    public static void toStop() {
        INSTANCE.stop();
    }

    /**
     * Triggers a job with the specified parameters.
     *
     * @param jobId                 The ID of the job to trigger.
     * @param triggerType           The type of trigger.
     * @param failRetryCount        The number of retry attempts.
     * @param executorShardingParam Sharding parameters for the executor.
     * @param executorParam         Parameters for the executor.
     */
    public static void trigger(int jobId, TriggerTypeEnum triggerType, int failRetryCount, String executorShardingParam, String executorParam) {
        INSTANCE.addTrigger(jobId, triggerType, failRetryCount, executorShardingParam, executorParam);
    }
}
